package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo16Student {

  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("student")

    val sc = new SparkContext(conf)

    val scoreRDD: RDD[String] = sc.textFile("data/score.txt")

    val courseRDD: RDD[String] = sc.textFile("data/course.txt")

    val scoresRDD: RDD[(String, (String, String))] = scoreRDD.map(score =>{

      val splits: Array[String] = score.split(",")

      //学生id
      val sid: String = splits(0)
      //课程id
      val cid: String = splits(1)
      //该科分数
      val sco: String = splits(2)

      (cid,(sid,sco))

    })

    val coursesRDD: RDD[(String, String)] = courseRDD.map(course =>{
      val splits: Array[String] = course.split(",")
      //课程id
      val cid: String = splits(0)
      //课程总分
      val sco: String = splits(2)
      (cid,sco)
    })

    val joinRDD: RDD[(String, ((String, String), String))] = scoresRDD.join(coursesRDD)

    //joinRDD.foreach(println)

    val s_2: RDD[(String, Double)] = joinRDD.map(joins => {
      //取分数
      val score: Double = joins._2._1._2.toDouble

      //取学号
      val sid: String = joins._2._1._1

      //取总分
      val cco: Double = joins._2._2.toDouble

      //归一化
      val pre: Double = score / cco

      //返回学号和各科的归一化数据，即：pre = 得分/总分
      (sid,pre)
    })

    //s_2.foreach(println)
    //根据学号分组,再map这个组
    val id_DX: RDD[(String, Double)] = s_2.groupBy(_._1).map(s => {
      //学号
      val id: String = s._1

      //学号+各科归一的比值的集合
      val id_pre: Iterable[(String, Double)] = s._2

      //单独取出各科归一的比值的集合
      val sumpre: Iterable[Double] = id_pre.map(_._2)

      //计算该学生学科平均值
      val avg: Double = sumpre.sum / sumpre.size

      //计算方差的分子和的各个部分
      val chazhi: Iterable[Double] = id_pre.map(s => (s._2-avg)*(s._2-avg))

      //得到方差值
      val DX: Double = chazhi.sum / chazhi.size

      //返回，学号和方差
      (id,DX)
    })

    //根据方差降序排序并打印
    id_DX.sortBy(-_._2).foreach(println)


  }

}
